package dev.devlok.springcloudstreamzeromqbinder;

import dev.devlok.springcloudstreamzeromqbinder.config.ZeroMqMessageBinderConfiguration;
import dev.devlok.springcloudstreamzeromqbinder.provisioners.ZeroMqMessageBinderProvisioner;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.context.Lifecycle;
import org.springframework.integration.core.MessageProducer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;

import static java.nio.file.StandardOpenOption.*;
import static org.assertj.core.api.Assertions.*;
import static org.junit.jupiter.api.Assertions.*;

@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = {ZeroMqMessageBinderConfiguration.class})
public class ZeroMqMessageBinderTests {

    private static final String INPUT_DESTINATION = "input";
    private static final String OUTPUT_DESTINATION = "output";
    @Autowired
    private ZeroMqMessageBinderProvisioner zeroMqMessageBinderProvisioner;
    @Autowired
    private ZeroMqMessageBinder fileMessageBinder;

    private ZContext zContext;
    private ZMQ.Socket socket;

    @BeforeEach
    public void setUp() {
        cleanUp(OUTPUT_DESTINATION, INPUT_DESTINATION);
        zContext = new ZContext();
    }

    @Test
    public void messageProducerShouldArchiveReadMessages() throws Exception {
        assertNotNull(zeroMqMessageBinderProvisioner);
        assertNotNull(fileMessageBinder);


        ConsumerProperties consumerProperties = new ConsumerProperties();
        ConsumerDestination consumerDestination =
                zeroMqMessageBinderProvisioner.provisionConsumerDestination(INPUT_DESTINATION, null, consumerProperties);


        MessageProducer messageProducer =
                fileMessageBinder.createConsumerEndpoint(consumerDestination, null, consumerProperties);
        ((Lifecycle) messageProducer).start();

        socket = zContext.createSocket(SocketType.PUB);
        socket.bind("tcp://localhost:6001");

        Thread.sleep(100);
        socket.sendMore("input");
        socket.send("ABC");

    }

    private void cleanUp(String... destinations) {

    }

}
